package aeemqx

import (
	"context"
	"devops-super/internal/consts"
	"devops-super/internal/model/entity"
	"devops-super/internal/model/mymodel"
	"devops-super/internal/service"
	"encoding/base64"
	"encoding/json"
	"errors"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/net/gclient"
	"github.com/gogf/gf/v2/os/gctx"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/util/gconv"
	"strconv"
	"strings"
)

type sAeemqx struct{}

func init() {
	service.RegisterAeemqx(New())
}

func New() *sAeemqx {
	return &sAeemqx{}
}

// DoEmqxSub 初始一个订阅mqtt的实例 专门用来接收消息
func (s *sAeemqx) DoEmqxSub(ctx context.Context, mqttConf *mymodel.MqttConnectOption) (err error) {
	obj := GetAeEmqxInstance()
	obj.NewAeMqttClient(ctx, mqttConf)
	return nil
}

// HandleAeMqttData 接收处理mqtt数据
func (s *sAeemqx) HandleAeMqttData(ctx context.Context, msg mqtt.Message) {
	topic := msg.Topic()
	//glog.Line().Info(ctx, topic)

	if s.IsDevicePubTopic(ctx, topic) || s.IsMobileAppPubTopic(ctx, topic) {
		payload := msg.Payload()
		go func(ctx context.Context, payload []byte) {
			err := s.ConsumeDevicePub(ctx, payload)
			if err != nil {
				glog.Line().Info(ctx, err.Error())
				return
			}
		}(ctx, payload)
		return
	}

	//emqx的上线通知
	if s.IsDeviceConnectTopic(ctx, topic) {
		arr := strings.Split(topic, "/")
		clientID := arr[len(arr)-2]
		err := service.Device().ChangeDeviceOnlineStatus(ctx, clientID, consts.DeviceStatusOnline)
		if err != nil {
			return
		}
		err = s.RecoverDeviceTmpInfo(ctx, clientID)
		if err != nil {
			return
		}
		return
	}
	//emqx的离线通知
	if s.IsDeviceDisConnectTopic(ctx, topic) {
		arr := strings.Split(topic, "/")
		clientID := arr[len(arr)-2]
		err := service.Device().ChangeDeviceOnlineStatus(ctx, clientID, consts.DeviceStatusOffline)
		if err != nil {
			return
		}
		return
	}
	// mochi-mqtt-server的上线通知
	if s.IsDeviceMochiConnectTopic(ctx, topic) {
		arr := strings.Split(topic, "/")
		clientID := arr[len(arr)-1]
		err := service.Device().ChangeDeviceOnlineStatus(ctx, clientID, consts.DeviceStatusOnline)
		if err != nil {
			return
		}
		err = s.RecoverDeviceTmpInfo(ctx, clientID)
		if err != nil {
			return
		}
		return
	}
	// mochi-mqtt-server的离线通知
	if s.IsDeviceMochiDisConnectTopic(ctx, topic) {
		arr := strings.Split(topic, "/")
		clientID := arr[len(arr)-1]
		err := service.Device().ChangeDeviceOnlineStatus(ctx, clientID, consts.DeviceStatusOffline)
		if err != nil {
			return
		}
		return
	}

}

// InnerCommonAePublish AE数据发给设备端
func (s *sAeemqx) InnerCommonAePublish(ctx context.Context, mqttConf *mymodel.MqttConnectOption, topic string, data []byte) (err error) {
	obj := GetAeEmqxInstance()
	err = obj.ClientSend(ctx, mqttConf, topic, 0, false, data)
	return err
}

// ClearRetainedMqttMsg 由于更新导致的 retained 消息需要清除
func (s *sAeemqx) ClearRetainedMqttMsg(ctx context.Context, mqttConf *mymodel.MqttConnectOption, topic string) (err error) {
	obj := GetAeEmqxInstance()
	a := ""
	err = obj.ClientSend(ctx, mqttConf, topic, 0, true, []byte(a))
	return err
}

// WebsiteMqttPublish 网站内部mqtt发送
func (s *sAeemqx) WebsiteMqttPublish(ctx context.Context, mqttConf *mymodel.MqttConnectOption, topic string, data []byte) (err error) {
	obj := GetAeEmqxInstance()
	err = obj.ClientSend(ctx, mqttConf, topic, 0, false, data)
	return err
}

// GetMqttConfByServerKey 根据mqtt的服务键 获取配置文件里的连接信息
func (s *sAeemqx) GetMqttConfByServerKey(ctx context.Context, serverKey string) (out *mymodel.MqttConnectOption, err error) {
	mqttConfigKey := consts.EnvConfig_Mqtt + "." + serverKey // 比如 mqtt.server1
	mqttConfString, err := g.Cfg().Get(ctx, mqttConfigKey)
	if err != nil {
		return nil, err
	}
	err = gconv.Structs(mqttConfString, &out)
	return
}

// CommonAePublish 通用的mqtt发送消息
// aeType 是记录发送的数据类型
func (s *sAeemqx) CommonAePublish(ctx context.Context, deviceNo string, aeType string, data []byte, ops *mymodel.EmqxPublishOps) (err error) {
	deviceInfo, err := service.Device().GetByDeviceNo(ctx, deviceNo)
	if err != nil {
		return err
	}
	if deviceInfo == nil || deviceInfo.Id == 0 {
		return errors.New("device not found")
	}
	realSendTo := deviceInfo
	var prow *entity.AppDevice
	if deviceInfo.ParentId > 0 { //子设备
		prow, err = service.Device().GetOne(ctx, int(deviceInfo.ParentId))
		if err != nil {
			return err
		}
		if prow == nil || prow.Id == 0 {
			return errors.New("parent device 1 not found")
		}
		// 跟送快递一样，协议定义终点收信人，但是中间转了几首，不再协议中指明，需要中间网关自己记录，
		if prow.ParentId > 0 { //如果套一层娃，说明是网关的网关，则再找一次父级。 最多支持套一次娃
			prow, err = service.Device().GetOne(ctx, int(deviceInfo.ParentId))
			if err != nil {
				return err
			}
			if prow == nil || prow.Id == 0 {
				return errors.New("parent device 2 not found")
			}
		}
		realSendTo = prow
	}
	if realSendTo.OpenStatus == consts.OpenStatusInactive {
		err = errors.New("device is inactive")
		return
	}
	// 检查设备的上线情况
	mqttconf, err := s.GetMqttConfByServerKey(ctx, realSendTo.MqttServer)
	if err != nil {
		return err
	}
	if mqttconf == nil {
		return errors.New("mqtconf is nil")
	}
	isOnline := false
	if mqttconf.Tag == consts.MqttTagEmqx { //只有是emqx4.4的时候才需要去调用此api
		isOnline, err = s.ApiClientIsOnline(ctx, mqttconf, realSendTo.DeviceNo)
		if err != nil {
			return err
		}
		if !isOnline {
			err := service.Device().ChangeDeviceOnlineStatus(ctx, realSendTo.DeviceNo, consts.DeviceStatusOffline)
			if err != nil {
				return err
			}
			if ops != nil && ops.MustOnline {
				err = errors.New("device is offline")
				return err
			}
			//如果设备不在线，如果发送的是retained为true的信息，保存，等到设备在线时，立马发送
			if ops != nil && ops.Retained {
				err := s.SaveDeviceTmpInfo(ctx, realSendTo.DeviceNo, aeType, data, ops)
				if err != nil {
					return err
				}
			}
			//不在线就不发了
			return nil
		}
	}
	if data == nil {
		data = []byte(aeType)
	}
	realSendToProductInfo, err := service.Product().GetByProductId(ctx, realSendTo.ProductId)
	if err != nil || realSendToProductInfo == nil {
		return errors.New("product not found")
	}
	realSendToProductKey := realSendToProductInfo.ProductKey
	if deviceInfo.Type == consts.ProductTypeGateway { //如果是网关
		if realSendToProductKey == consts.ProductKeyQCGW {
			topic := consts.EMQX_TOPIC_QC_GW_SUB + realSendTo.DeviceNo
			err = s.InnerCommonAePublish(ctx, mqttconf, topic, data)
		} else if realSendToProductKey == consts.ProductKeyMCM1 { //如果是AEMC 则跟清诚网关类似发送，但是topic为普通设备topic
			topic := consts.EMQX_TOPIC_SUB + realSendTo.DeviceNo
			err = s.InnerCommonAePublish(ctx, mqttconf, topic, data)
		}
	} else if deviceInfo.Type == consts.ProductTypeSub { //网关子设备
		if realSendToProductKey == consts.ProductKeyQCGW {
			//如果是清诚自研网关，则走直接发送protobuf不分包发送方式
			topic := consts.EMQX_TOPIC_QC_GW_SUB + realSendTo.DeviceNo
			err = s.InnerCommonAePublish(ctx, mqttconf, topic, data)
		} else if realSendToProductKey == consts.ProductKeyMCM1 { //如果是AEMC 则跟清诚网关类似发送，但是topic为普通设备topic
			topic := consts.EMQX_TOPIC_SUB + realSendTo.DeviceNo
			err = s.InnerCommonAePublish(ctx, mqttconf, topic, data)
		}
	} else {
		// ae直连设备的网关topic
		topic := consts.EMQX_TOPIC_SUB + deviceInfo.DeviceNo
		err = s.InnerCommonAePublish(ctx, mqttconf, topic, data)
		if err != nil {
			return
		}
	}
	return
}

// ApiClientIsOnline 只有emqx4.4此接口调用才有作用
func (s *sAeemqx) ApiClientIsOnline(ctx context.Context, mqttConf *mymodel.MqttConnectOption, clientId string) (result bool, err error) {
	path := "/api/v4/clients/" + clientId
	url, _ := s.GetApiBaseUrl(ctx, mqttConf, path)
	auth, _ := s.GetApiAuth(ctx, mqttConf)
	c := g.Client()
	c.SetHeader("Authorization", auth)
	resp, e := c.Get(gctx.New(), url)
	defer func(resp *gclient.Response) {
		err := resp.Close()
		if err != nil {
			return
		}
	}(resp)
	if e != nil {
		err = errors.New("request mqtt api server error")
	} else {
		respStr := resp.ReadAllString()
		var resdata *mymodel.EmqxApiClientOnlineResult
		err := json.Unmarshal([]byte(respStr), &resdata)
		if err != nil {
			return false, err
		} else {
			if resdata.Code == 0 && len(resdata.Data) > 0 && resdata.Data[0].Connected {
				return true, nil
			} else {
				return false, nil
			}
		}
	}
	return false, err
}

// GetApiBaseUrl 只有emqx4.4此接口调用才有作用
func (s *sAeemqx) GetApiBaseUrl(ctx context.Context, mqttConf *mymodel.MqttConnectOption, path string) (res string, err error) {
	if mqttConf == nil {
		return "", errors.New("mqtconf is nil")
	}
	protocol := "http"
	if mqttConf.Protocol == "https" {
		protocol = mqttConf.Protocol
	}
	res = protocol + "://" + mqttConf.Host + ":" + strconv.Itoa(mqttConf.ApiPort) + path
	return
}

// GetApiAuth 只有emqx4.4此接口调用才有作用
func (s *sAeemqx) GetApiAuth(ctx context.Context, mqttConf *mymodel.MqttConnectOption) (res string, err error) {
	if mqttConf == nil {
		return "", errors.New("mqtconf is nil")
	}
	str := mqttConf.ApiUser + ":" + mqttConf.ApiPassword
	res = "Basic " + base64.StdEncoding.EncodeToString([]byte(str))
	return
}
